package dacp.etl.kafka.hdfs.connect;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.errors.ConnectException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.codahale.metrics.Slf4jReporter;

import dacp.etl.kafka.hdfs.connect.monitor.MRegistry;
import io.confluent.common.config.ConfigException;
import io.confluent.connect.hdfs.HdfsSinkConnectorConfig;
import io.confluent.connect.hdfs.Version;

public class HdfsSinkConnector extends Connector {

	private static final Logger log = LoggerFactory.getLogger(HdfsSinkConnector.class);
	private Map<String, String> configProperties;
	private HdfsSinkConnectorConfig config;

	@Override
	public ConfigDef config() {
		return HdfsSinkConnectorConfig.getConfig();
	}

	@Override
	public void start(Map<String, String> props) {
		try {
			configProperties = props;
			config = new HdfsSinkConnectorConfig(props);
			
			final Slf4jReporter reporter = Slf4jReporter.forRegistry(MRegistry.get())
	                .outputTo(LoggerFactory.getLogger("dacp.etl.kafka.hdfs.v1.Monitor"))
	                .convertRatesTo(TimeUnit.SECONDS)
	                .convertDurationsTo(TimeUnit.MILLISECONDS)
	                .build();
			reporter.start(60, TimeUnit.SECONDS); 
			
		} catch (ConfigException e) {
			throw new ConnectException("Couldn't start HdfsSinkConnector due to configuration error", e);
		}
	}

	@Override
	public void stop() {
		// TODO Auto-generated method stub
	}

	@Override
	public Class<? extends Task> taskClass() {
		return HdfsSinkTask.class;
	}

	@Override
	public List<Map<String, String>> taskConfigs(int maxTasks) {
		List<Map<String, String>> taskConfigs = new ArrayList<>();
		
		for (int i = 0; i < maxTasks; i++) {
			Map<String, String> taskProps = new HashMap<>();
			taskProps.putAll(configProperties);
			taskProps.put("task.id", i+"");
			taskConfigs.add(taskProps);
		}

		System.setProperty("java.security.krb5.conf", "/home/etl/dacp-realstream-server-1.0.0/conf/krb5.conf");
		System.setProperty("java.security.auth.login.config", "/home/etl/dacp-realstream-server-1.0.0/conf/kafka-jaas.conf");
		return taskConfigs;
	}

	@Override
	public String version() {
		return Version.getVersion();
	}

}
